package com.ttdevs.net.observer.observer.tcp;

import java.util.concurrent.BlockingQueue;

import com.ttdevs.net.observer.NetObservable;
import com.ttdevs.net.observer.NetSubstrate;
import com.ttdevs.net.observer.dao.ExcepitonPacket;
import com.ttdevs.net.observer.exception.HeadDataErrorException;
import com.ttdevs.net.observer.observer.DataReceiveObserver;

public class TCPTransferDateRunnable implements Runnable {

	private BlockingQueue<byte[]> receiveDataQueue;
	private DataReceiveObserver dataReceiveObserver;
	private NetObservable netObservable;

	private boolean isRun = false;

	public TCPTransferDateRunnable(BlockingQueue<byte[]> receiveDataQueue,
			DataReceiveObserver dataReceiveObserver, NetObservable netObservable) {
		this.receiveDataQueue = receiveDataQueue;
		this.dataReceiveObserver = dataReceiveObserver;
		this.netObservable = netObservable;

		isRun = true;
	}

	public final void stop() {
		isRun = false;
		Thread.currentThread().interrupt();
	}

	@Override
	public void run() {
		byte[] data = null;
		try {
			while (isRun) {
				data = receiveDataQueue.take();
				if (checkDataValidity(data)) {
					dataReceiveObserver.update(netObservable, data);
				} else {
					HeadDataErrorException headError = new HeadDataErrorException(
							"tcp packet head is not 0xaa55");
					netObservable.notifyObservers(new ExcepitonPacket(
							NetSubstrate.E_HEADDATAERROR, headError));
				}

			}
		} catch (InterruptedException e) {
			netObservable.notifyObservers(new ExcepitonPacket(
					NetSubstrate.E_INTERRUPTED, e));
		}
	}

	private boolean checkDataValidity(byte[] data) {
		if (data[0] == (byte) 0xaa && data[1] == (byte) 0x55) {
			return true;
		} else {
			return false;
		}
	}
}
